package com.insight.common.message.websocket;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.yuanqiao.insight.common.util.common.RedisUtils;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import com.insight.common.constant.CacheConstant;
import com.insight.common.message.websocket.factory.WebSocketHandlerFactory;
import com.insight.common.message.websocket.handler.IWebSocketHandler;
import com.insight.common.message.webssh.handler.WebSSHHandler;
import com.insight.common.mq.RedisMqInit;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.websocket.OnClose;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * @Author scott
 * @Date 2019/11/29 9:41
 * @Description: 此注解相当于设置访问URL
 */
@Component
@Slf4j
@ServerEndpoint("/websocket/{userId}") //此注解相当于设置访问URL
public class WebSocket {

    private static WebSocketHandlerFactory webSocketHandlerFactory;

    private Session session;

    private IWebSocketHandler webSocketHandler;

    private static final CopyOnWriteArraySet<WebSocket> webSockets = new CopyOnWriteArraySet<>();

    private static final Map<String, Map<String, Session>> userSessions = new ConcurrentHashMap<>();

    @Getter
    private static final Map<String, Session> sessionPool = new HashMap<>();

    @Autowired
    public void setWebSocketHandlerFactory(WebSocketHandlerFactory webSocketHandlerFactory) {
        WebSocket.webSocketHandlerFactory = webSocketHandlerFactory;
    }

    @OnOpen
    public void onOpen(Session session, @PathParam(value = "userId") String userId) {
        //System.out.println("      ------------------------ WebSocket.onOpen() -------------------------------     ");
        try {
            if (session == null || userId == null || userId.trim().isEmpty()) {
                log.warn("无效连接参数，session 或 userId 为空");
                if (session != null) {
                    session.close();
                }
                return;
            }

            this.session = session;
            webSockets.add(this);
            sessionPool.put(userId, session);
            userSessions.computeIfAbsent(userId, k -> new ConcurrentHashMap<>()).put(session.getId(), session);

            RedisUtils redisUtils = RedisMqInit.getRedisUtils();
            if (redisUtils != null) {
                redisUtils.hset(CacheConstant.SYS_WS_CACHE, userId, RedisMqInit.getCONSUMER());
            }

            if (userId.contains("_")) {
                String[] split = userId.split("_");
                if (split.length >= 2) { // 至少有两个部分
                    String code = split[0];
                    String messageType = split[1];
                    webSocketHandler = webSocketHandlerFactory.getWebSocketHandler(messageType);
                    if (webSocketHandler != null) {
                        webSocketHandler.onOpenHandler(code, session);
                    }
                }
            }

            log.info("【websocket消息】有新的连接，总数为: {}", webSockets.size());
        } catch (Exception e) {
            log.error("WebSocket onOpen 异常", e);
        }
    }

    @OnClose
    public void onClose(Session session, @PathParam("userId") String userId) {
        //System.out.println("      ------------------------ WebSocket.onClose() -------------------------------     ");
        try {
            if (webSocketHandler != null) {
                webSocketHandler.onCloseHandler(this.session);
            }
            webSockets.remove(this);
            log.info("【websocket消息】连接断开，总数为: {}", webSockets.size());
        } catch (Exception e) {
            log.error("WebSocket onClose 异常（前半部分）", e);
        }

        // 清理用户连接映射
        try {
            if (userId != null && userSessions.containsKey(userId)) {
                Map<String, Session> userConnections = userSessions.get(userId);
                if (userConnections != null && session != null) {
                    userConnections.remove(session.getId());
                    if (userConnections.isEmpty()) {
                        userSessions.remove(userId);
                    }
                }
            }
        } catch (Exception e) {
            log.error("清理用户连接映射失败", e);
        }
    }

    @OnMessage
    public void onMessage(String message, Session session) {
        //接收到前端心跳检测后向前端发送消息
        if (StringUtils.equals("HeartBeat", message)) {
            if (session != null) {
                session.getAsyncRemote().sendText(message);
            }
        }

        if (webSocketHandler != null) {
            try {
                webSocketHandler.onMessageHandler(message, this.session);
            } catch (RuntimeException e) {
                log.error("WebSocket消息处理运行时异常", e);
                try {
                    session.getBasicRemote().sendText("系统异常：" + e.getMessage());
                    webSocketHandler.onCloseHandler(this.session);
                    this.session.close();
                } catch (IOException ex) {
                    log.error("关闭WebSocket会话失败", ex);
                }
            } catch (Exception e) {
                log.error("WebSocket消息处理检查异常", e);
                try {
                    session.getBasicRemote().sendText("系统异常：" + e.getMessage());
                    webSocketHandler.onCloseHandler(this.session);
                    this.session.close();
                } catch (IOException ex) {
                    log.error("关闭WebSocket会话失败", ex);
                }
            }
        }
    }

    // 此为广播消息
    public static void sendAllMessage(String message) {
        log.info("【WebSocket消息】广播消息：{}", message);
        for (WebSocket webSocket : webSockets) {
            if (webSocket.webSocketHandler instanceof WebSSHHandler) {
                continue;
            }
            try {
                if (webSocket != null && webSocket.session != null && webSocket.session.isOpen()) {
                    synchronized (webSocket.session) {
                        webSocket.session.getBasicRemote().sendText(message != null ? message : "");
                    }
                }
            } catch (Exception e) {
                log.error("广播消息失败", e);
            }
        }
    }


    public static void sendMessage(Session session, String message) {
        try {
            log.info("【websocket消息】 单点消息: {}", message);

            if (session == null) return;

            session.getAsyncRemote().sendText(message);
        } catch (Exception e) {
            log.error("WebSocket消息发送异常", e);
        }
    }

    // 此为单点消息
    public static void sendOneMessage(String userId, String message) {
        Session session = sessionPool.get(userId);
        if (session != null && session.isOpen()) {
            sendMessage(session, message);
        } else {
            final RedisUtils redisUtils = RedisMqInit.getRedisUtils();
            if (redisUtils.hHasKey(CacheConstant.SYS_WS_CACHE, userId)) {
                final String key = String.valueOf(RedisMqInit.getRedisUtils().hget(CacheConstant.SYS_WS_CACHE, userId));
                JSONObject data = new JSONObject();
                data.put("userId", userId);
                data.put("message", getMessage(message));
                RedisMqInit.getRedisMq()
                        .publish(CacheConstant.STREAM_PREFIX + "websocket:" + key, data);
            }
        }
    }

    // 此为单点消息(多人)
    public static void sendMoreMessage(String[] userIds, String message) {
        for (String userId : userIds) {
            sendOneMessage(userId, message);
        }

    }

    // 此为单点消息(多人)
    public static void sendMoreMessageByUserMap(String[] userIds, String message) {
        for (String userId : userIds) {
            sendOneMessageByUserMap(userId, message);
        }

    }

    //此为单点消息通过Map<String, Map<String, Session>> 发送消息
    public static void sendOneMessageByUserMap(String userId, String message) {
        Map<String, Session> userConnections = userSessions.get(userId);
        if (userConnections != null) {
            for (Session userSession : userConnections.values()) {
                try {
                    if (userSession != null && userSession.isOpen()) {
                        userSession.getAsyncRemote().sendText(message != null ? message : "");
                    }
                } catch (Exception e) {
                    log.error("发送消息失败", e);
                }
            }
        } else {
            final RedisUtils redisUtils = RedisMqInit.getRedisUtils();
            if (redisUtils != null && redisUtils.hHasKey(CacheConstant.SYS_WS_CACHE, userId)) {
                final String key = String.valueOf(redisUtils.hget(CacheConstant.SYS_WS_CACHE, userId));
                JSONObject data = new JSONObject();
                data.put("userId", userId);
                data.put("message", getMessage(message));
                try {
                    RedisMqInit.getRedisMq().publish(CacheConstant.STREAM_PREFIX + "websocket:" + key, data);
                } catch (Exception e) {
                    log.error("Redis 消息发布失败", e);
                }
            }
        }
    }

    private static Object getMessage(String message) {
        // 如果 message 为 null 或空字符串，直接返回原值，避免解析异常
        if (message == null || message.trim().isEmpty()) {
            return message;
        }

        try {
            return JSON.parseObject(message);
        } catch (Exception e) {
            // 解析失败，返回原始字符串（可能是纯文本消息）
            log.debug("消息无法解析为 JSON，将作为普通字符串返回: {}", message);
            return message;
        }
    }

}
